package com.shujia.streaming

import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.DStream

object Demo04Transform {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo04Transform")
      .master("local[2]")
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    val ssc: StreamingContext = new StreamingContext(spark.sparkContext, Durations.seconds(5))


    val lineDS: DStream[String] = ssc.socketTextStream("master", 8888)
    // 实际上SparkStreaming中每一次触发的job最终是转换成RDD完成的
    // 所以可以将实时计算 转换成 基于RDD的计算
    val ds: DStream[Row] = lineDS
      // 将每个批次的数据转换成RDD进行处理
      // 类似的还有foreachRDD，一个有返回值可以再进行后续的处理 一个没有返回值
      .transform(rdd => {
        // 将RDD转换成DF
        // 统计每个批次数据中的班级人数
        val stuDF: DataFrame = rdd.map(line => {
          val splits: Array[String] = line.split(",")
          (splits(0), splits(1), splits(2).toInt, splits(3), splits(4))
        }).toDF("id", "name", "age", "gender", "clazz")
        val clazzCntDF: DataFrame = stuDF
          .groupBy($"clazz")
          .agg(count("*") as "cnt")
        clazzCntDF.rdd
      })

    ds.filter(_.getAs[String]("clazz") != "文科一班")
      .print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
